#include "server.h"


typedef int (*P_CMD_FUNC)(char *, char *);
#define CMD_FUNC(_svc)   \
int CMD_FUNC_##_svc(char *request, char *respond)

static unsigned int quit=IS_FALSE;

static struct
{/*{{{*/
    SKT skt;
    MSGQ_SYN msgq;
    char *request;
    char *respond;
}/*}}}*/
monitor_ctl;

static CMD_FUNC(PRINTTHREAD);
static CMD_FUNC(PRINTSERVER);
static CMD_FUNC(PRINTSERVICE);
static CMD_FUNC(DOWNSVR);
static CMD_FUNC(DOWNALL);
static CMD_FUNC(HELP);

static struct _cmd_map
{/*{{{*/
    char *name;
    char *ab;
    P_CMD_FUNC pfunc;
    char *desc;
    char *usage;
}/*}}}*/
cmd_map[]=
{/*{{{*/
    {
        "printthread", "pthrd", CMD_FUNC_PRINTTHREAD,
        "print thread connections & calls",
        "printthread(pthrd)"
    },
    {
        "printserver", "psvr", CMD_FUNC_PRINTSERVER,
        "print server calls",
        "printserver(psvr) [server]"
    },
    {
        "printservice", "psvc", CMD_FUNC_PRINTSERVICE,
        "print service calls",
        "printservice(psvc) [service]"
    },
    {
        "downsvr", "", CMD_FUNC_DOWNSVR,
        "shutdown server",
        "downsvr {server}"
    },
    {
        "downall", "", CMD_FUNC_DOWNALL,
        "shutdown whole svcenv",
        "downall"
    },
    {
        "help", "", CMD_FUNC_HELP,
        "",
        ""
    }
};/*}}}*/
static const unsigned int CMD_MAP_SIZE=
sizeof(cmd_map)/sizeof(struct _cmd_map);

static CMD_FUNC(PRINTTHREAD)
{/*{{{*/
    int offset=0;
    unsigned int conn_num=0;
    unsigned int client_count=0;
    unsigned int client_call=0;
    int worker_idx;

    offset+=sprintf(respond+offset,
            "Thread  |  Queued Conns | Served Conns | Calls\n");
    offset+=sprintf(respond+offset,
            "------     ------------   ------------   ----------\n");
    pthread_mutex_lock(&stat_info._worker_.lock);
    for(worker_idx=0; worker_idx<cnfg_info.threads; ++worker_idx)
    {
        struct _worker_info *worker=
            &stat_info._worker_.worker[worker_idx];
        offset+=sprintf(respond+offset,
                "   %3d             %4d       %8d     %8d\n",
                worker_idx,
                worker->conn_num,
                worker->client_count,
                worker->client_call);
        conn_num+=worker->conn_num;
        client_count+=worker->client_count;
        client_call+=worker->client_call;
    }
    pthread_mutex_unlock(&stat_info._worker_.lock);
    offset+=sprintf(respond+offset,
            "------     ------------   ------------   ----------\n"
            "Total:            %5d     %10d   %10d\n",
            conn_num, client_count, client_call);

    return(offset);
}/*}}}*/

static CMD_FUNC(PRINTSERVER)
{/*{{{*/
    char svrname[NAME_SIZE+1]="";
    int offset=0;

    if(request)
    {
        char *p=strchr(request, SPACE);
        int svr_size;
        if(p)
            svr_size=p-request;
        else
            svr_size=strlen(request);
        if(svr_size > NAME_SIZE)
        {
            offset+=sprintf(respond+offset, "Invalid server name\n");
            return(offset);
        }
        strncpy(svrname, request, svr_size);
        svrname[svr_size]=ZERO;
    }

    struct _svr_info *svr;
    int svr_idx;
    struct listp_link *p;
    int svr_call;

    pthread_mutex_lock(&stat_info._svr_.lock);
    if(BUF_IS_CLR(svrname))
    {/*{{{*/
        offset+=sprintf(respond+offset,
                "Server                 | Calls\n");
        offset+=sprintf(respond+offset,
                "------                   -----\n");
        for(svr_idx=0; svr_idx<SVR_MAX_NUM; ++svr_idx)
        {
            svr=&stat_info._svr_.svrs[svr_idx];
            if(svr->pid != -1)
            {
                p=&svr->svcs;
                svr_call=0;
                while((p=p->next) != &svr->svcs)
                {
                    struct _svc_rec *svc=listp_entry(p,
                            struct _svc_rec, link_svr);
                    svr_call+=svc->call;
                }
                offset+=sprintf(respond+offset,
                        "%-20s     %d\n",
                        svr->name, svr_call);
            }
        }
    }/*}}}*/
    else
    {/*{{{*/
        for(svr_idx=0; svr_idx<SVR_MAX_NUM; ++svr_idx)
        {
            svr=&stat_info._svr_.svrs[svr_idx];
            if(strcasecmp(svr->name, svrname) == 0)
            {
                p=&svr->svcs;
                svr_call=0;

                offset+=sprintf(respond+offset,
                        "Service                    | Calls\n");
                offset+=sprintf(respond+offset,
                        "-------                      -----\n");
                while((p=p->next) != &svr->svcs)
                {
                    struct _svc_rec *svc=listp_entry(p,
                            struct _svc_rec, link_svr);
                    offset+=sprintf(respond+offset,
                            "%-20s     %d\n",
                            svc->name, svc->call);
                    svr_call+=svc->call;
                }
                offset+=sprintf(respond+offset,
                        "------\n"
                        "Total:                       %d",
                        svr_call);
                break;
            }
        }
        if(svr_idx == SVR_MAX_NUM)
        {
            offset+=sprintf(respond+offset,
                    "Server(%s) not found\n", svrname);
        }
    }/*}}}*/
    pthread_mutex_unlock(&stat_info._svr_.lock);

    return(offset);
}/*}}}*/

static CMD_FUNC(PRINTSERVICE)
{/*{{{*/
    char svcname[NAME_SIZE+1]="";
    int offset=0;

    if(request)
    {
        char *p=strchr(request, SPACE);
        int size;
        if(p)
            size=p-request;
        else
            size=strlen(request);
        if(size > NAME_SIZE)
        {
            offset+=sprintf(respond+offset, "Invalid service name\n");
            goto GO_OUT;
        }
        strncpy(svcname, request, size);
        svcname[size]=ZERO;
    }

    struct _svr_info *svr;
    struct _svc_rec *svc;
    pthread_mutex_lock(&stat_info._svr_.lock);
    if(BUF_IS_CLR(svcname))
    {/*{{{*/
        int svr_idx;
        offset+=sprintf(respond+offset,
                "Service                    | Server          | Calls\n");
        offset+=sprintf(respond+offset,
                "-------                      -----             -----\n");
        for(svr_idx=0; svr_idx<SVR_MAX_NUM; ++svr_idx)
        {
            svr=&stat_info._svr_.svrs[svr_idx];
            struct listp_link *p=&svr->svcs;
            while((p=p->next) != &svr->svcs)
            {
                svc=listp_entry(p, struct _svc_rec,
                        link_svr);
                offset+=sprintf(respond+offset,
                        "%-20s         %-15s   %d\n",
                        svc->name, svr->name, svc->call);
            }
        }
    }/*}}}*/
    else
    {/*{{{*/
        struct rb_link *p;
        p=rb_find(&stat_info._svr_.svcs, svcname);
        if(p)
        {
            svc=tree_entry(p, struct _svc_rec, link_svc);
            offset+=sprintf(respond+offset,
                    "Service                    | Server          | Calls\n");
            offset+=sprintf(respond+offset,
                    "-------                      -----             -----\n");
            offset+=sprintf(respond+offset,
                    "%-20s         %-15s   %d\n",
                    svc->name,
                    stat_info._svr_.svrs[svc->svr_idx].name,
                    svc->call);
        }
        else
        {
            offset+=sprintf(respond+offset,
                    "Service(%s) not found\n", svcname);
        }
    }/*}}}*/
    pthread_mutex_unlock(&stat_info._svr_.lock);

GO_OUT:
    return(offset);
}/*}}}*/

static CMD_FUNC(DOWNSVR)
{/*{{{*/
    int offset=0;
    char buf[200];
    int res;

    if(!request)
    {
        sprintf(buf, "Server name cannot be empty");
        errlog("%s: %s", __func__, buf);
        offset+=sprintf(respond+offset, "%s\n", buf);
        return(offset);
    }

    char svrname[NAME_SIZE+1];
    char *p=strchr(request, SPACE);
    int svr_size;
    if(p)
        svr_size=p-request;
    else
        svr_size=strlen(request);
    if(svr_size > NAME_SIZE)
    {
        sprintf(buf, "Server name too long");
        errlog("%s: %s, len(%d)", __func__, buf, svr_size);
        offset+=sprintf(respond+offset, "%s\n", buf);
        return(offset);
    }
    strncpy(svrname, request, svr_size);
    svrname[svr_size]=ZERO;

    pthread_mutex_lock(&stat_info._svr_.lock);

    int i;
    for(i=0; i<SVR_MAX_NUM; ++i)
    {
        struct _svr_info *svr=&stat_info._svr_.svrs[i];
        if(strcmp(svr->name, svrname) == 0)
        {
            if(svr->pid != -1)
            {
                // 通知svr
                char queue[NAME_SIZE+1];
                sprintf(queue, "se_svr_%s_%d",
                        svrname, cnfg_info.ipckey);
                CLR_BUF(buf);
                para_put_item(buf, "SYS");
                para_put_item(buf, "DOWNSVR");
                int size=strlen(buf);
                res=msgq_syn_call(monitor_ctl.msgq, queue,
                        (unsigned char *)buf, size,
                        (unsigned char *)buf, &size,
                        2*1000);
                if(res)
                {
                    errlog("%s: call svr(%s) fail, res:%d",
                            __func__, svr->name);
                    offset+=sprintf(respond+offset,
                            "Server(%s) down fail, try again\n",
                            svr->name);
                }
                else
                {
                    // 清除svr信息
                    while(!listp_is_empty(&svr->svcs))
                    {
                        struct _svc_rec *svc=
                            listp_entry(svr->svcs.next, struct _svc_rec,
                                    link_svr);
                        if(!rb_delete_link(&stat_info._svr_.svcs,
                                    &svc->link_svc))
                        {
                            errlog("%s: delete svc(%s) from "
                                    "stat_info._svr_.svcs fail, "
                                    "impossible",
                                    __func__, svc->name);
                        }
                        listp_del(&svc->link_svr);
                        free(svc);
                    }
                    svr->pid=-1;

                    sprintf(buf, "Server(%s) down succeed", svr->name);
                    errlog("%s: %s", __func__, buf);
                    offset+=sprintf(respond+offset, "%s\n", buf);
                }
            }
            else
            {
                sprintf(buf, "Server(%s) already down", svr->name);
                errlog("%s: %s", __func__, buf);
                offset+=sprintf(respond+offset, "%s\n", buf);
            }
            break;
        }
    }
    if(i== SVR_MAX_NUM)
    {
        sprintf(buf, "Server(%s) not found", svrname);
        errlog("%s: %s", __func__, buf);
        offset+=sprintf(respond+offset, "%s\n", buf);
    }

    pthread_mutex_unlock(&stat_info._svr_.lock);

    return(offset);
}/*}}}*/

static CMD_FUNC(DOWNALL)
{/*{{{*/
    int offset=0;
    char buf[200];
    int res;
    int i;

    /*--- 线程 ---*/
    static int TRY_NUM=10;

    pthread_mutex_lock(&operate_info._listener_.lock);
    operate_info._listener_.quit=IS_TRUE;
    pthread_mutex_unlock(&operate_info._listener_.lock);

    pthread_mutex_lock(&operate_info._worker_.lock);
    operate_info._worker_.quit=IS_TRUE;
    pthread_mutex_unlock(&operate_info._worker_.lock);

    sleep(1);

    for(i=0; i<TRY_NUM; ++i)
    {
        int worker_idx;

        pthread_mutex_lock(&operate_info._listener_.lock);
        if(operate_info._listener_.state == THRD_RUNNING)
        {
            pthread_mutex_unlock(&operate_info._listener_.lock);
            goto CONTINUE;
        }
        pthread_mutex_unlock(&operate_info._listener_.lock);

        pthread_mutex_lock(&operate_info._worker_.lock);
        for(worker_idx=0; worker_idx<cnfg_info.threads; ++worker_idx)
        {
            if(operate_info._worker_.state[worker_idx] == THRD_RUNNING)
            {
                pthread_mutex_unlock(&operate_info._worker_.lock);
                goto CONTINUE;
            }
        }
        pthread_mutex_unlock(&operate_info._worker_.lock);

        break;

CONTINUE:
        sleep(1);
    }
    if(i == TRY_NUM)   // 未关闭
    {
        sprintf(buf, "Down agent not complete, try again");
        errlog("%s: %s", __func__, buf);
        offset+=sprintf(respond+offset, "%s\n", buf);
        return(offset);
    }

    /*--- 服务 ---*/
    pthread_mutex_lock(&stat_info._svr_.lock);

    for(i=0; i<SVR_MAX_NUM; ++i)
    {
        struct _svr_info *svr=&stat_info._svr_.svrs[i];
        if(svr->pid != -1)
        {
            char queue[NAME_SIZE+1];
            sprintf(queue, "se_svr_%s_%d", svr->name, cnfg_info.ipckey);
            CLR_BUF(buf);
            para_put_item(buf, "SYS");
            para_put_item(buf, "DOWNSVR");
            int size=strlen(buf);
            res=msgq_syn_call(monitor_ctl.msgq, queue,
                    (unsigned char *)buf, size,
                    (unsigned char *)buf, &size,
                    5*1000);
            if(res)
            {
                errlog("%s: call svr(%s) fail, res:%d",
                        __func__, svr->name, res);
                offset+=sprintf(respond+offset,
                        "Server(%s) down fail, try again\n", svr->name);
                goto GO_OUT;
            }
            sprintf(buf, "Server(%s) down succeed", svr->name);
            errlog("%s: %s", __func__, buf);
            offset+=sprintf(respond+offset, "%s\n", buf);
        }
    }

    // 全部关闭
    offset+=sprintf(respond+offset,
            "Shutdown complete");
    quit=IS_TRUE;

GO_OUT:
    pthread_mutex_unlock(&stat_info._svr_.lock);

    return(offset);
}/*}}}*/

static CMD_FUNC(HELP)
{/*{{{*/
    struct _cmd_map *map=NULL;
    int offset=0;
    int i;

    // 所有帮助
    if(request)
        while(*request == SPACE)
            request++;
    if(!request || BUF_IS_CLR(request))
    {
        offset+=sprintf(respond+offset, "help [command]\n");
        offset+=sprintf(respond+offset, "commands:\n");
        for(i=1; i<CMD_MAP_SIZE; ++i)
        {
            offset+=sprintf(respond+offset, "    %s",
                    cmd_map[i].name);
            if(!BUF_IS_CLR(cmd_map[i].ab))
                offset+=sprintf(respond+offset, "(%s)",
                        cmd_map[i].ab);
            offset+=sprintf(respond+offset, "\n");
        }
        return(offset);
    }

    // 单独命令帮助
    for(i=0; i<CMD_MAP_SIZE-1; ++i)
    {
        map=&cmd_map[i];
        if(strcasecmp(map->name, request) == 0 ||
                strcasecmp(map->ab, request) == 0)
        {
            offset+=sprintf(respond+offset, "Description: %s\n",
                    map->desc);
            offset+=sprintf(respond+offset, "Usage: %s\n",
                    map->usage);
            return(offset);
        }
    }

    offset+=sprintf(respond+offset, "Command not found, type help\n");
    return(offset);
}/*}}}*/

static int _callcmd(char *request, char *respond)
{/*{{{*/
    int offset=0;
    struct _cmd_map *map=NULL;
    char *body;
    int svcsize;

    body=strchr(request, SPACE);
    if(body)
    {
        svcsize=body-request;
        while(*body == SPACE) body++;
    }
    else
        svcsize=strlen(request);
    if(svcsize > 0 && svcsize <= NAME_SIZE)
    {
        char svc[NAME_SIZE+1];
        int i;

        strncpy(svc, request, svcsize);
        svc[svcsize]=ZERO; 
        for(i=0; i<CMD_MAP_SIZE; ++i)
        {
            map=&cmd_map[i];
            if(strcasecmp(map->name, svc) == 0 ||
                    strcasecmp(map->ab, svc) == 0)
                break;
        }
        if(i == CMD_MAP_SIZE)
        {
            offset+=sprintf(respond+offset, "command(%s) not found",
                    svc);
            errlog("%s: %s", __func__, respond);
            offset+=sprintf(respond+offset, "\n");
        }
        else
        {
            offset+=map->pfunc(body, respond);
        }
    }
    else
    {
        offset+=sprintf(respond+offset, "command invalid");
        errlog("%s: %s", __func__, respond);
        offset+=sprintf(respond+offset, "\n");
    }
    offset+=sprintf(respond+offset, TEXTEND);

    return(offset);
}/*}}}*/

int monitor_start()
{/*{{{*/
    int IDLE_TIMEOUT=SECOND_OF_MINUTE(5);
    int TIMEOUT=5;
    int TIMEOUT_MS=(TIMEOUT*1000);
    char buf[200];
    int res;

    char queue[NAME_SIZE+1];
    sprintf(queue, "se_agent_monitor_%d", cnfg_info.ipckey); 
    res=msgq_clear(queue);
    if(res)
    {
        sprintf(buf, "agent: clear queue(%s) fail, res:%d",
                queue, res);
        fprintf(stderr, "%s\n", buf); errlog(buf);
        return(-1);
    }
    res=msgq_syn_attach(queue, &monitor_ctl.msgq);
    if(res)
    {
        sprintf(buf, "agent: msgq_syn_attach(%s) fail, res:%d",
                queue, res);
        fprintf(stderr, "%s\n", buf); errlog(buf);
        return(-1);
    }

    ALLOC_MULTI_DOERR(monitor_ctl.request, char, BUFFER_SIZE,
            sprintf(buf, "agent: alloc monitor_ctl.request fail");
            fprintf(stderr, "%s\n", buf); errlog(buf);
            return(-1);
            );
    char *request=monitor_ctl.request;
    int request_size;
    ALLOC_MULTI_DOERR(monitor_ctl.respond, char, SOCK_MSG_SIZE,
            sprintf(buf, "agent: alloc monitor_ctl.respond fail");
            fprintf(stderr, "%s\n", buf); errlog(buf);
            return(-1);
            );
    char *respond=monitor_ctl.respond;
    int respond_size;

    char addr[LONG_FILE_NM_SIZE+1];
    sprintf(addr, "%d", cnfg_info.monitor_port);
    SKT term_skt;
    res=skt_listen(addr, 0, &monitor_ctl.skt);
    if(res)
    {
        fprintf(stderr, "agent: open monitor port fail\n");
        return(-1);
    }
    fprintf(stderr, "agent: open monitor port(%s)\n", addr);

    int timeout_ms;
    int accu_timeout;
    while(1)
    {
        timeout_ms=TIMEOUT_MS;
        res=skt_accept(monitor_ctl.skt, &timeout_ms, &term_skt);
        if(res)
        {
            if(res != RET_ERR_TIMEOUT)
                errlog("%s: accept fail", __func__);
            continue;
        }

        accu_timeout=0;
        while(1)
        {
            timeout_ms=TIMEOUT_MS;
            res=skt_read(term_skt,
                    (unsigned char *)request, &request_size,
                    SKT_R_LINE, &timeout_ms);
            if(res < 0)
            {
                if(res == RET_ERR_TIMEOUT)
                {
                    accu_timeout+=TIMEOUT;
                    if(accu_timeout < IDLE_TIMEOUT)
                        continue;
                    errlog("%s: console timeout, revoke", __func__);
                }
                else
                {
                    if(res == RET_ERR_ABORT)
                        errlog("%s: console revoke", __func__);
                    else
                        errlog("%s: read console fail, res:%d, revoke",
                                __func__, res);
                }
                skt_close(term_skt);
                break;
            }
            if(request[request_size-1] == RTN)
            {
                if(request[request_size-2] == '\r')
                    request[request_size-2]=ZERO;
                else
                    request[request_size-1]=ZERO;
            }
            else
                request[request_size]=ZERO;
            trim_head_space(request);
            trim_tail_space(request);
            accu_timeout=0;

            respond_size=_callcmd(request, respond);

            timeout_ms=TIMEOUT_MS;
            res=skt_write(term_skt,
                    (unsigned char *)respond, &respond_size,
                    SKT_W_SIZE, &timeout_ms);
            if(res < 0)
            {
                errlog("%s: respond to console fail, res:%d, revoke",
                        __func__, res);
                skt_close(term_skt);
                break;
            }

            // 退出
            if(quit)
            {
                sleep(1);
                skt_close(term_skt);
                goto QUIT;
            }
        } // console
    }

QUIT:
    return(0);
}/*}}}*/
